/*
 * Copyright (c) 2006-2007 Terracotta, Inc. All rights reserved.
 */
package org.terracotta.workmanager.pipemanager;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.terracotta.workmanager.pipemanager.PipeManager;
import org.terracotta.workmanager.routing.RoutableWorkItem;

/**
 * A pipe manager that manages the different pipes.
 */
public class PipeManager<ID> {

  public static enum Initiator {
    MASTER, WORKER, OBSERVER
  }

  private Initiator type;

  private Pipe.Factory<RoutableWorkItem<ID>, ID> pipeFactory;

  private final Map<ID, Pipe<RoutableWorkItem<ID>, ID>> workerPipes =
      new ConcurrentHashMap<ID, Pipe<RoutableWorkItem<ID>, ID>>();

  private final Map<ID, Integer> workersOs = new ConcurrentHashMap<ID, Integer>();

  private transient Set<RoutableWorkItem<ID>> allCompletedWork =
      new HashSet<RoutableWorkItem<ID>>();

  private volatile boolean isRunning = true;

  public PipeManager(final Pipe.Factory<RoutableWorkItem<ID>, ID> factory, final ID nodeID,
      Initiator type, int os) {
    System.out.println("created PipeManager for Initiator: " + type);
    pipeFactory = factory;
    this.type = type;

    if (type == Initiator.WORKER) {
      synchronized (workersOs) {
        workersOs.put(nodeID, os);
      }
    }
  }

  /**
   * Gets the Pipe associated with a routing ID
   *
   * @param routingID
   * @return the pipe associated with the routingID, null if not found
   */
  public Pipe<RoutableWorkItem<ID>, ID> get(final ID routingID) {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    if (routingID == null) throw new IllegalArgumentException("routing ID is null");

    synchronized (workerPipes) {
      Pipe<RoutableWorkItem<ID>, ID> pipe;
      if (workerPipes.containsKey(routingID)) {
        return workerPipes.get(routingID);
      } else {
        return null;
      }

    }
  }

  public Set<RoutableWorkItem<ID>> getAllCompletedWork() {
    return allCompletedWork;
  }

  public Pipe<RoutableWorkItem<ID>, ID> getOrCreatePipeFor(final ID routingID) {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    if (routingID == null) throw new IllegalArgumentException("routing ID is null");

    synchronized (workerPipes) {
      Pipe<RoutableWorkItem<ID>, ID> pipe;
      if (workerPipes.containsKey(routingID)) {
        pipe = workerPipes.get(routingID);
      } else {
        pipe = pipeFactory.create(routingID);
        workerPipes.put(routingID, pipe);
        workerPipes.notifyAll();
        System.out.println("registering a new work queue for routing ID: " + routingID);
      }
      switch (type) {
        case MASTER:
          pipe.startQueueListener(allCompletedWork);
          break;
        case WORKER:
          break;
      }
      return pipe;
    }
  }

  public Map<ID, Pipe<RoutableWorkItem<ID>, ID>> getPipes() {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    return workerPipes;
  }

  public Map<ID, Pipe<RoutableWorkItem<ID>, ID>> getPipesForOs(int os) {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");

    Map<ID, Pipe<RoutableWorkItem<ID>, ID>> pipes =
        new ConcurrentHashMap<ID, Pipe<RoutableWorkItem<ID>, ID>>();

    for (Entry<ID, Integer> worker : workersOs.entrySet()) {
      if (worker.getValue() == os) {
        if (workerPipes.get(worker.getKey()) != null) {
          pipes.put(worker.getKey(), workerPipes.get(worker.getKey()));
        }
      }
    }

    return pipes;
  }

  public List<ID> getRoutingIDs() {
    return Collections.unmodifiableList(new ArrayList<ID>(workerPipes.keySet()));
  }

  public int getWorkerOs(final ID nodeID) {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    if (nodeID == null) throw new IllegalArgumentException("node ID is null");

    synchronized (workersOs) {
      if (!workersOs.containsKey(nodeID)) {
        return -1;
      } else {
        return workersOs.get(nodeID);
      }
    }
  }

  public Collection<ID> getWorkers() {
    return workersOs.keySet();
  }

  public boolean hasEmptyPipe() {
    for (Pipe<RoutableWorkItem<ID>, ID> pipe : workerPipes.values()) {
      if (pipe.getPendingWorkQueue().size() == 0) {
        return true;
      }
    }
    return false;
  }

  public boolean hasEmptyPipeForOs(int os) {
    for (Pipe<RoutableWorkItem<ID>, ID> pipe : getPipesForOs(os).values()) {
      if (pipe.getPendingWorkQueue().size() == 0) {
        return true;
      }
    }
    return false;
  }

  public void put(final RoutableWorkItem<ID> workItem, final ID routingID)
      throws InterruptedException {

    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    Pipe<RoutableWorkItem<ID>, ID> pipe = getOrCreatePipeFor(routingID);
    pipe.getPendingWorkQueue().put(workItem); // blocks if queue is full
  }

  public void removePipeFor(final ID routingID) {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    if (routingID == null) throw new IllegalArgumentException("routing ID is null");

    synchronized (workersOs) {
      if (workersOs.containsKey(routingID)) {
        workersOs.remove(routingID);
      }
    }

    synchronized (workerPipes) {
      if (!workerPipes.containsKey(routingID)) {
        return;
      }

      System.out.println("Clearing the pipe for routing ID: " + routingID);
      workerPipes.get(routingID).clear();
    }
    System.out.println("Removing the pipe for routing ID: " + routingID);
    workerPipes.remove(routingID);
    System.out.println("Done");
  }

  public void reset() {
    allCompletedWork.clear();

    synchronized (workerPipes) {
      isRunning = false;
      for (Pipe<RoutableWorkItem<ID>, ID> pipe : workerPipes.values()) {
        pipe.clear();
      }
    }
    isRunning = true;
  }

  public void sendCommand(ID routingID, String command) throws InterruptedException {
    if (!isRunning) throw new IllegalStateException("pipe manager has been shut down");
    Pipe<RoutableWorkItem<ID>, ID> pipe = getOrCreatePipeFor(routingID);
    pipe.getCommandQueue().put(command); // blocks if queue is full
  }
}
